package org.example.zookeeper.test;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;
import org.example.zookeeper.config.ZkProperties;

import java.util.List;

public class CuratorTester implements ZookeeperTester {
    @Override
    public void test(ZkProperties zkConfig) {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient(zkConfig.getServer(), zkConfig.getSessionTimeout(), zkConfig.getConnectionTimeout(), retryPolicy);
        client.start();
        try {
            String path = "/zk-curator";
            deleteNode(client, path);
            deleteNode(client, path);
            // IllegalArgumentException: Path must start with / character
            // createNode(client, "start", CreateMode.PERSISTENT);
            createNode(client, path, CreateMode.PERSISTENT);
            createNode(client, path + "/c1", CreateMode.PERSISTENT);
            createNode(client, path + "/c1/c11", CreateMode.PERSISTENT);
            createNode(client, path + "/c1/c12", CreateMode.PERSISTENT);
            createNode(client, path + "/e1", CreateMode.EPHEMERAL);

            getChildren(client, path);
            getChildren(client, path + "/temp");
            getChildren(client, path + "/c1");

            getChildrenInBackground(client, path);

            getDate(client, path + "/c1");
            getDate(client, path + "/c2");

            setData(client, path + "/c1", null);
            setData(client, path + "/c2", "hello".getBytes());

            getDate(client, path + "/c1");

            print("fin");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (client != null) {
                client.close();
            }
        }
    }

    private void deleteNode(CuratorFramework client, String path) throws Exception {
        //KeeperException$NotEmptyException
        try {
            client.delete().deletingChildrenIfNeeded().forPath(path);
        } catch (KeeperException.NoNodeException e) {
            print("delete node error: " + e.getMessage());
        }
    }

    private void createNode(CuratorFramework client, String path, CreateMode mode) throws Exception {
        Stat stat = client.checkExists().forPath(path);
        if (stat == null) {
            String v = client.create().withMode(mode).forPath(path);
            print("created " + v);
        } else {
            print(stat);
        }
    }

    private void getDate(CuratorFramework client, String path) throws Exception {
        try {
            byte[] bytes = client.getData().usingWatcher(new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    System.out.println("get data watcher: " + watchedEvent);
                }
            }).forPath(path);
            if (bytes != null) {
                print(new String(bytes));
            } else {
                print("not data found");
            }
        } catch (KeeperException.NoNodeException e) {
            print("node not found");
        }
    }

    private void setData(CuratorFramework client, String path, byte[] data) throws Exception {
        try {
            Stat stat = client.setData().forPath(path, data);
            if (stat != null) {
                print(stat);
            } else {
                print("set data failed");
            }
        } catch (KeeperException.NoNodeException e) {
            print("[warning] node not found");
        }
    }

    private void getChildren(CuratorFramework client, String path) throws Exception {
        try {
            List<String> children = client.getChildren().usingWatcher(new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    System.out.println("get children watcher: " + watchedEvent);
                }
            }).forPath(path);
            print("children size: " + children.size() + " children is: " + children);
        } catch (KeeperException.NoNodeException e) {
            print("[warning] parent node not found");
        }
    }

    private void getChildrenInBackground(CuratorFramework client, String path) throws Exception {
        List<String> children = client.getChildren().inBackground((CuratorFramework client1, CuratorEvent event1) -> {
            System.out.println("background children on event: " + event1.getStat());
            System.out.println("background children on event: " + event1.getChildren());
        }).withUnhandledErrorListener((String message, Throwable e) -> {
            System.out.println("background children on error: " + message);
            e.printStackTrace();
        }).forPath(path);
        System.out.println("background children immediately: " + children);
        Thread.sleep(500);
    }
}
